package simpledb.storage;

import simpledb.common.Database;
import simpledb.common.Permissions;
import simpledb.common.DbException;
import simpledb.common.DeadlockException;
import simpledb.execution.Predicate;
import simpledb.lock.LockManager;
import simpledb.lock.LockType;
import simpledb.transaction.TransactionAbortedException;
import simpledb.transaction.TransactionId;

import javax.print.attribute.standard.ReferenceUriSchemesSupported;
import java.io.*;

import java.sql.SQLOutput;
import java.sql.Time;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * BufferPool manages the reading and writing of pages into memory from
 * disk. Access methods call into it to retrieve pages, and it fetches
 * pages from the appropriate location.
 * <p>
 * The BufferPool is also responsible for locking;  when a transaction fetches
 * a page, BufferPool checks that the transaction has the appropriate
 * locks to read/write the page.
 *
 * @Threadsafe, all fields are final
 */
public class BufferPool {

    // 新增显式锁保护整个链表操作（包括delete+add的组合操作）
    private final ReentrantLock lruLock = new ReentrantLock();

    private static class Node{
        PageId key;
        Page value;
        Node prev, next;

        public Node() {
            this(null, null);
        }

        public Node(PageId key, Page value) {
            this.key = key;
            this.value = value;
        }
    }

    private void delete(Node node) {
        Node prev = node.prev;
        Node next = node.next;
        prev.next = next;
        next.prev = prev;
        node.prev = null;
        node.next = null;
    }

    private void add(Node node) {
        Node next = head.next;
        head.next = node;
        next.prev = node;
        node.next = next;
        node.prev = head;
    }

    private final Node head;
    private final Node tail;


    /** Bytes per page, including header. */
    private static final int DEFAULT_PAGE_SIZE = 4096;

    private static int pageSize = DEFAULT_PAGE_SIZE;

    /** Default number of pages passed to the constructor. This is used by
     other classes. BufferPool should use the numPages argument to the
     constructor instead. */
    public static final int DEFAULT_PAGES = 50;

    private final Map<PageId, Node> pageCache;
    private final int maxPages;

    private final LockManager lockManager;

    private final Random random = new Random(System.currentTimeMillis());

    /**
     * Creates a BufferPool that caches up to numPages pages.
     *
     * @param numPages maximum number of pages in this buffer pool.
     */
    public BufferPool(int numPages) {
        // some code goes here
        pageCache = new ConcurrentHashMap<>();
        maxPages = numPages;
        head = new Node();
        tail = new Node();
        head.next = tail;
        tail.prev = head;
        // this.lockManager = new LockManager();
        // this.lockManager = new LockManager(true, 10, this);
        this.lockManager = new LockManager(false, 50, this);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // System.out.println("[BufferPool] 初始化缓冲区，最大缓存页数：" + numPages);
    }

    // 提供获取内部锁的方法（供外部使用）
    public Lock getCustomLock() {
        return lruLock; // 暴露锁对象，外部可通过此方法获取
    }

    // 事务中止标记
    private final Set<TransactionId> abortedTransactions = ConcurrentHashMap.newKeySet();

    // 标记事务为中止（死锁检测器调用）
    public void markTransactionAborted(TransactionId tid) {
        abortedTransactions.add(tid);
    }

    // 检查事务是否已中止（线程安全）
    public boolean isTransactionAborted(TransactionId tid) {
        return abortedTransactions.contains(tid);
    }


    public static int getPageSize() {
        return pageSize;
    }

    // THIS FUNCTION SHOULD ONLY BE USED FOR TESTING!!
    public static void setPageSize(int pageSize) {
        BufferPool.pageSize = pageSize;
        // System.out.println("[BufferPool] 测试模式：设置页面大小为 " + pageSize + " 字节");
    }

    // THIS FUNCTION SHOULD ONLY BE USED FOR TESTING!!
    public static void resetPageSize() {
        BufferPool.pageSize = DEFAULT_PAGE_SIZE;
        // System.out.println("[BufferPool] 测试模式：重置页面大小为默认 " + DEFAULT_PAGE_SIZE + " 字节");
    }

    /**
     * Retrieve the specified page with the associated permissions.
     * Will acquire a lock and may block if that lock is held by another
     * transaction.
     * <p>
     * The retrieved page should be looked up in the buffer pool.  If it
     * is present, it should be returned.  If it is not present, it should
     * be added to the buffer pool and returned.  If there is insufficient
     * space in the buffer pool, a page should be evicted and the new page
     * should be added in its place.
     *
     * @param tid the ID of the transaction requesting the page
     * @param pid the ID of the requested page
     * @param perm the requested permissions on the page
     */
    public  Page getPage(TransactionId tid, PageId pid, Permissions perm)
            throws TransactionAbortedException, DbException {
        // System.out.println("[BufferPool] 事务 " + tid + " 请求页面 " + pid + "，权限：" + perm);

        LockType lockType;
        if(perm == Permissions.READ_ONLY){
            lockType = LockType.SHARE_LOCK;
            // System.out.printf("[TID" + tid.getId() + "] 节点ID-%d尝试获取读锁\n", pid.getPageNumber());
        }else {
            lockType = LockType.EXCLUSIVE_LOCK;
            // System.out.printf("[TID" + tid.getId() + "] 节点ID-%d尝试获取写锁\n", pid.getPageNumber());
        }

        try {
            if (!lockManager.acquireLock(pid, tid, lockType, 10+random.nextInt(10), 2+ random.nextInt(2))){
                System.out.println("获取锁失败");
                throw new TransactionAbortedException("获取锁发生异常， 可能死锁");
            }
        } catch (TransactionAbortedException | InterruptedException e) {
            // 获取锁失败，回滚事务
            System.out.printf("[TID" + tid.getId() + "] 节点ID-%d获取锁失败，回滚事务\n", pid.getPageNumber());
            // e.printStackTrace();
            throw new TransactionAbortedException("获取锁发生异常， 可能死锁");
        }

        if(perm == Permissions.READ_ONLY){
            // System.out.printf("[TID" + tid.getId() + "] 节点ID-%d尝试获取读锁成功\n", pid.getPageNumber());
        }else {
            // System.out.printf("[TID" + tid.getId() + "] 节点ID-%d尝试获取写锁\n", pid.getPageNumber());
        }

        try {
            lruLock.lock();
            Node node = pageCache.get(pid);
            if (node != null) {
                if (lockType.equals(LockType.EXCLUSIVE_LOCK)) {
                    node.value.markDirty(true, tid);
                    System.out.printf("[TID" + tid.getId() + "] 1 节点ID-%d获取读写锁,并标记为脏页\n", pid.getPageNumber());
                }
                // 缓存命中：移动节点到头部（维护LRU）
                delete(node);
                add(node);
                return node.value;
            }
            // 缓存未命中：先释放锁，避免IO操作阻塞锁
            // lruLock.unlock();

            // 磁盘IO操作（耗时，不持有锁）
            Page page = Database.getCatalog().getDatabaseFile(pid.getTableId()).readPage(pid);

            // 重新获取锁，处理缓存添加
            // lruLock.lock();
            // 二次检查：防止释放锁期间其他线程已添加该页面
            Node existing = pageCache.get(pid);
            if (existing != null) {
                delete(existing);
                add(existing);
                if (lockType.equals(LockType.EXCLUSIVE_LOCK)) {
                    existing.value.markDirty(true, tid);
                    System.out.printf("[TID" + tid.getId() + "] 2 节点ID-%d获取读写锁,并标记为脏页\n", pid.getPageNumber());
                }
                return existing.value;
            }
            // 缓存满则淘汰页面
            if (pageCache.size() >= maxPages) {
                evictPage();
            }
            // 添加新页面到缓存
            Node newNode = new Node(pid, page);
            add(newNode);
            pageCache.put(pid, newNode);
            if (lockType.equals(LockType.EXCLUSIVE_LOCK)) {
                newNode.value.markDirty(true, tid);
                System.out.println(newNode.value.isDirty());
                System.out.printf("[TID" + tid.getId() + "] 3 节点ID-%d获取读写锁,并标记为脏页\n", pid.getPageNumber());
            }
            return page;
        } finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    /**
     * Releases the lock on a page.
     * Calling this is very risky, and may result in wrong behavior. Think hard
     * about who needs to call this and why, and why they can run the risk of
     * calling it.
     *
     * @param tid the ID of the transaction requesting the unlock
     * @param pid the ID of the page to unlock
     */
    public  void unsafeReleasePage(TransactionId tid, PageId pid) {
        // 底层会做检查，所以没必要互斥
        // System.out.println("不安全释放锁");
        try {
            lruLock.lock();
            flushPage(pid, true);
            lockManager.releasePage(tid, pid);
            delete(pageCache.get(pid));
            pageCache.remove(pid);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    /**
     * Release all locks associated with a given transaction.
     *
     * @param tid the ID of the transaction requesting the unlock
     */
    public void transactionComplete(TransactionId tid) {
        // System.out.println("[BufferPool] 事务 " + tid + " 完成，释放所有锁");
        transactionComplete(tid,true);
    }

    /** Return true if the specified transaction has a lock on the specified page */
    public boolean holdsLock(TransactionId tid, PageId p) {
        // some code goes here
        // not necessary for lab1|lab2
        return false;
    }

    private void rollBack(TransactionId tid) {
        System.out.println("[TID" + tid.getId() + "], rollBack start");
        try {
            lruLock.lock();
            for (Node value : pageCache.values()) {
                Page page = value.value;
                TransactionId dirty = page.isDirty();
                if (tid.equals(dirty)) {
                    System.out.println("[TID" + tid.getId() + "]" + "回滚页面 " + page.getId().getPageNumber());
                    int tableId = page.getId().getTableId();
                    DbFile file = Database.getCatalog().getDatabaseFile(tableId);
                    value.value = file.readPage(page.getId());
                    delete(value);
                    add(value);
                    page.markDirty(false, null);
                }else if (dirty!=null) {
                    System.out.printf("[TID" + tid.getId() + "]" + "回滚页面时节点ID-%d的事务为%d不匹配，跳过\n", page.getId().getPageNumber(), dirty.getId());
                }
            }
        }finally {
            System.out.println("[TID" + tid.getId() + "], rollBack end");
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    /**
     * Commit or abort a given transaction; release all locks associated to
     * the transaction.
     *
     * @param tid the ID of the transaction requesting the unlock
     * @param commit a flag indicating whether we should commit or abort
     */
    public void transactionComplete(TransactionId tid, boolean commit) {
        try {
            lruLock.lock();
            if (commit) {
                System.out.println("[transactionComplete] [TID" + tid.getId() + "]" + "提交事务 ");
                flushPages(tid);
            } else {
                System.out.println("[transactionComplete] [TID" + tid.getId() + "]" + "回滚事务 ");
                // 标记事务为中止
                markTransactionAborted(tid);
                rollBack(tid);
            }
        } catch (IOException e) {
            System.out.println("[BufferPool] 事务 " + tid + " 提交时刷新失败：" + e.getMessage());
            // e.printStackTrace();
            throw new RuntimeException(e);
        } finally {
            // 确保锁一定会释放
            System.out.println("[transactionComplete] [TID" + tid.getId() + "]" + "释放锁 ");
            lockManager.releaseAllLocks(tid);
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    // 更新脏页到缓存
    private void updateCache(List<Page> dirtyPages, TransactionId tid) {
        try {
            lruLock.lock();
            for (Page dirtyPage : dirtyPages) {

                dirtyPage.markDirty(true, tid);
                if (!pageCache.containsKey(dirtyPage.getId())) {
                    // 缓存满则淘汰
                    if (pageCache.size() >= maxPages) {
                        try {
                            evictPage();
                        } catch (DbException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    // 添加到缓存
                    Node node = new Node(dirtyPage.getId(), dirtyPage);
                    add(node);
                    pageCache.put(dirtyPage.getId(), node);
                }
            }
        } finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }
    // updateCache中添加节点也需要加锁
    // private void updateCache(List<Page> dirtyPages, TransactionId tid) {
    //     lruLock.lock();
    //     for (Page dirtyPage : dirtyPages) {
    //         // 原有逻辑不变...
    //         if (!pageCache.containsKey(dirtyPage.getId())) {
    //             // ...可能的evictPage调用（已加锁）
    //             Node node = new Node(dirtyPage.getId(), dirtyPage);
    //             add(node);
    //             pageCache.put(dirtyPage.getId(), node);
    //         }
    //     }
    //     lruLock.unlock();
    // }

    /**
     * Add a tuple to the specified table on behalf of transaction tid.  Will
     * acquire a write lock on the page the tuple is added to and any other
     * pages that are updated (Lock acquisition is not needed for lab2).
     * May block if the lock(s) cannot be acquired.
     *
     * Marks any pages that were dirtied by the operation as dirty by calling
     * their markDirty bit, and adds versions of any pages that have
     * been dirtied to the cache (replacing any existing versions of those pages) so
     * that future requests see up-to-date pages.
     *
     * @param tid the transaction adding the tuple
     * @param tableId the table to add the tuple to
     * @param t the tuple to add
     */
    public void insertTuple(TransactionId tid, int tableId, Tuple t)
            throws DbException, IOException, TransactionAbortedException {
        // System.out.println("[transactionComplete] [TID" + tid  + " 从页面 " + t.getRecordId().getPageId() + t);
        DbFile file =  Database.getCatalog().getDatabaseFile(tableId);
        List<Page> pages = file.insertTuple(tid, t);
        updateCache(pages, tid);
    }

    /**
     * Remove the specified tuple from the buffer pool.
     * Will acquire a write lock on the page the tuple is removed from and any
     * other pages that are updated. May block if the lock(s) cannot be acquired.
     *
     * Marks any pages that were dirtied by the operation as dirty by calling
     * their markDirty bit, and adds versions of any pages that have
     * been dirtied to the cache (replacing any existing versions of those pages) so
     * that future requests see up-to-date pages.
     *
     * @param tid the transaction deleting the tuple.
     * @param t the tuple to delete
     */
    public  void deleteTuple(TransactionId tid, Tuple t)
            throws DbException, IOException, TransactionAbortedException {
        DbFile file = Database.getCatalog().getDatabaseFile(t.getRecordId().getPageId().getTableId());
        List<Page> pages = file.deleteTuple(tid, t);
        updateCache(pages, tid);
    }

    /**
     * Flush all dirty pages to disk.
     * NB: Be careful using this routine -- it writes dirty data to disk so will
     *     break simpledb if running in NO STEAL mode.
     *     刷全部意味着未提交的事务页也会被刷盘，NO STEAL mode不允许这种操作
     */
    public void flushAllPages() throws IOException {
        System.out.println("[BufferPool] 开始刷新所有脏页到磁盘");
        Set<PageId> set = Set.copyOf(pageCache.keySet());
        for (PageId pageId : set) {
            flushPage(pageId, false);
        }
    }

    /** Remove the specific page id from the buffer pool.
     Needed by the recovery manager to ensure that the
     buffer pool doesn't keep a rolled back page in its
     cache.

     Also used by B+ tree files to ensure that deleted pages
     are removed from the cache so they can be reused safely
     */
    // public synchronized void discardPage(PageId pid) {
    //     // System.out.println("[BufferPool] 从缓存中丢弃页面 " + pid);
    //     Node node = pageCache.get(pid);
    //     if (node != null) {
    //         delete(node);
    //         pageCache.remove(pid);
    //         // System.out.println("[BufferPool] 页面 " + pid + " 已从缓存中移除，当前缓存页数：" + pageCache.size());
    //     } else {
    //         // System.out.println("[BufferPool] 页面 " + pid + " 不在缓存中，无需丢弃");
    //     }
    // }
    // discardPage中操作链表也需要加锁
    public void discardPage(PageId pid) {
        try {
            lruLock.lock();
            Node node = pageCache.get(pid);
            if (node != null) {
                delete(node);
                pageCache.remove(pid);
            }
        } finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    /**
     * Flushes a certain page to disk
     * @param pid an ID indicating the page to flush
     *  mark用于标记是否要修改脏页标记,因为flushallpages的时候不应该清空脏页标记
     */
    private void flushPage(PageId pid, boolean clean) throws IOException {
        try {
            lruLock.lock();
            Node node = this.pageCache.get(pid);
            if (node!=null && node.value.isDirty()!=null) {
                System.out.println("写回页面 " + pid.getPageNumber());
                // System.out.println(node.value.isSlotUsed(0));
                // 先写日志,再写磁盘, WAL
                // 注意,日志文件的内部操作都是要获取锁的,bufferpool也持有锁,必须保证任何需要访问 BufferPool 的 LogFile 操作，必须先获取 BufferPool 的锁，再获取 LogFile 自身的锁
                // 通过规定访问锁的顺序,才能避免死锁.
                Database.getLogFile().logWrite(node.value.isDirty(), node.value.getBeforeImage(), node.value);
                Database.getCatalog().getDatabaseFile(pid.getTableId()).writePage(node.value);
                if (clean) {
                    node.value.markDirty(false, null);
                }
            }
        }finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    /** Write all pages of the specified transaction to disk.
     */
    public void flushPages(TransactionId tid) throws IOException {
        try {
            lruLock.lock();
            for (Map.Entry<PageId, Node> entry : pageCache.entrySet()) {
                Page page = entry.getValue().value;
                if (tid.equals(page.isDirty())) {
                    System.out.println("[TID" + tid.getId() + "]" + "准备写回页面 " + page.getId().getPageNumber());

                    // 不能在flushPage中调用保存前快照,因为flushPage可能被单独调用,此时会把没有提交的页面的状态更新为前快照,导致其他事务回滚时看到错误状态
                    // 只有事务提交刷新之后,更新前页才是合理的前快照
                    page.setBeforeImage();
                    flushPage(page.getId(), true);
                }
            }
        }finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

    // 只移除脏页,不会影响日志机制
    private void evictPage() throws DbException {
        try {
            lruLock.lock();
            Node cur = tail;
            while (cur.prev != head) {
                cur = cur.prev;
                // System.out.println(cur.key);
                // System.out.println(cur.value.getId().getPageNumber());
                // System.out.println(cur.value.isDirty());
                if (cur.value.isDirty() == null) {
                    delete(cur);
                    pageCache.remove(cur.key);
                    return;
                }
            }
            throw new DbException("All Page Are Dirty Page");
        } finally {
            if (lruLock.isHeldByCurrentThread()) {
                lruLock.unlock();
            }
        }
    }

}